package com.yanggu.flink.datastream_api.checkpoint

import org.apache.flink.contrib.streaming.state.{EmbeddedRocksDBStateBackend, RocksDBStateBackend}
import org.apache.flink.runtime.state.filesystem.FsStateBackend
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend
import org.apache.flink.runtime.state.memory.MemoryStateBackend
import org.apache.flink.runtime.state.storage.{FileSystemCheckpointStorage, JobManagerCheckpointStorage}
import org.apache.flink.streaming.api.scala._

object StateBackendDemo {

  def main(args: Array[String]): Unit = {
    //1. 流处理环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val hdfsPath = "hdfs://namenode:xxx/xxx"

    //关于新版本状态后端的配置可以见
    //https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/ops/state/state_backends/#migrating-from-legacy-backends
    //1. MemoryStateBackend等价于使用HashMapStateBackend和JobManagerCheckpointStorage
    env.setStateBackend(new MemoryStateBackend())
    env.setStateBackend(new HashMapStateBackend())
    env.getCheckpointConfig.setCheckpointStorage(new JobManagerCheckpointStorage())

    //2. FsStateBackend等价于使用HashMapStateBackend和FileSystemCheckpointStorage
    env.setStateBackend(new FsStateBackend(hdfsPath))
    env.setStateBackend(new HashMapStateBackend())
    env.getCheckpointConfig.setCheckpointStorage(new FileSystemCheckpointStorage(hdfsPath))

    //3. RocksDBStateBackend等价于使用EmbeddedRocksDBStateBackend和FileSystemCheckpointStorage
    env.setStateBackend(new RocksDBStateBackend(hdfsPath))
    env.setStateBackend(new EmbeddedRocksDBStateBackend(true))
    env.getCheckpointConfig.setCheckpointStorage(new FileSystemCheckpointStorage(hdfsPath))
  }

}
